package com.epoint.flinkdemo.state;

import com.epoint.flinkdemo.entity.SenSorReading;
import com.epoint.flinkdemo.map.SensorReadingMapFunction;
import com.epoint.flinkdemo.process.AlarmProcessFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import scala.Tuple3;

public class KeyedStateTest {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> source = env.socketTextStream("localhost", 7777);

        DataStream<SenSorReading> mapSource = source.map(new SensorReadingMapFunction());

        DataStream<Tuple3<String, Double, Double>> alarmSource = mapSource.keyBy("id").process(new AlarmProcessFunction(10.0));
        alarmSource.print();
        env.execute("key state test");
    }
}
